package initialize

import (
	"context"
	"encoding/json"
	"fmt"
	"go_final/mxshop_srvs/order_srv/global"
	"go_final/mxshop_srvs/order_srv/model"
	"os"
	"time"

	"github.com/apache/rocketmq-client-go/v2/producer"

	"gorm.io/gorm"

	"go.uber.org/zap"

	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
)

func InitRocketmqListener() {
	c, _ := rocketmq.NewPushConsumer(
		consumer.WithNameServer([]string{"120.24.67.46:9876"}),
		consumer.WithGroupName("mxshop-order"),
	)
	err := c.Subscribe(
		"order_timeout",
		consumer.MessageSelector{},
		orderTimeout,
	)

	if err != nil {
		panic("订阅MQ消息失败")
	}

	err = c.Start()
	if err != nil {
		fmt.Println("err", err)
		panic("开始订阅MQ消息失败")
	}
}

// orderTimeout 监听到归还订单消息以后调用归还库存接口
func orderTimeout(ctx context.Context,
	msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {

	for i := range msgs {
		fmt.Printf("收到消息咯....: %v \n", msgs[i])
		// 这个接口应该确保幂等性, 保证一个订单的库存不能多次归还, 没有扣减的库存不能归还
		// 新建一张表, 这个表记录了详细的订单扣减细节, 以及归还细节
		var orderInfo model.OrderInfo
		err := json.Unmarshal(msgs[i].Body, &orderInfo)
		if err != nil {
			zap.S().Errorf("解析json失败: %s", err.Error())
			return consumer.ConsumeSuccess, nil
		}

		fmt.Printf("获取到订单超时消息: %v \n", time.Now())
		// 查询订单支付状态, 如果未支付, 就归还库存, 修改订单状态
		var order model.OrderInfo
		res := global.DB.Model(model.OrderInfo{}).Where(model.OrderInfo{
			OrderSn: orderInfo.OrderSn,
		}).First(&order)
		if res.Error != nil && res.Error != gorm.ErrRecordNotFound {
			// 数据库出错
			return consumer.ConsumeRetryLater, nil
		}
		if res.RowsAffected == 0 {
			// 没有订单记录, 什么都不用做
			return consumer.ConsumeSuccess, nil
		}
		if order.Status != "TRADE_SUCCESS" {
			tx := global.DB.Begin()
			// 修改订单的状态为已关闭
			order.Status = "TRADE_CLOSED"
			tx.Save(&order)

			// 发送一条order_reback消息归还库存
			p, _ := rocketmq.NewProducer(producer.WithNameServer([]string{"120.24.67.46:9876"}))
			err := p.Start()

			if err != nil {
				tx.Rollback()
				fmt.Printf("生成producer失败: %s", err.Error())
				os.Exit(1)
			}

			msg := &primitive.Message{
				Topic: "order_reback",
				Body:  msgs[i].Body,
			}
			res, err := p.SendSync(context.Background(), msg)
			if err != nil {
				tx.Rollback()
				fmt.Printf("发送失败 %s\n", res.String())
				return consumer.ConsumeRetryLater, nil
			}

			tx.Commit()
		}
	}

	return consumer.ConsumeSuccess, nil
}
